package com.example.kafka.service;

import lombok.RequiredArgsConstructor;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

import java.time.Duration;
import java.util.Collections;

/**
 * 演示先提交位点后处理消息的错误示例
 */
@Service
@RequiredArgsConstructor
public class OffsetCommitBeforeProcessService {
    private final Logger log = LoggerFactory.getLogger(getClass());

    private final KafkaConsumer<String, String> consumer;

    /**
     * 用例1: 订单处理 - 先提交位点后处理订单
     * 错误示例：在处理订单前就提交了位点
     */
    public String processOrder(String orderData) {
        try {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                // 错误：先提交位点
                consumer.commitSync(Collections.singletonMap(
                        new TopicPartition(record.topic(), record.partition()),
                        new org.apache.kafka.clients.consumer.OffsetAndMetadata(record.offset() + 1)
                ));

                // 后处理订单
                processOrderLogic(record.value());
            }
            return "Order processed";
        } catch (Exception e) {
            log.error("Error processing order", e);
            throw new RuntimeException("Failed to process order", e);
        }
    }

    /**
     * 用例2: 支付处理 - 先提交位点后处理支付
     * 错误示例：在处理支付前就提交了位点
     */
    public String processPayment(String paymentData) {
        try {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                // 错误：先提交位点
                consumer.commitSync();

                // 后处理支付
                processPaymentLogic(record.value());
            }
            return "Payment processed";
        } catch (Exception e) {
            log.error("Error processing payment", e);
            throw new RuntimeException("Failed to process payment", e);
        }
    }

    /**
     * 用例3: 库存更新 - 先提交位点后更新库存
     * 错误示例：在更新库存前就提交了位点
     */
    public String updateInventory(String inventoryData) {
        try {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                // 错误：先提交位点
                consumer.commitAsync();

                // 后更新库存
                updateInventoryLogic(record.value());
            }
            return "Inventory updated";
        } catch (Exception e) {
            log.error("Error updating inventory", e);
            throw new RuntimeException("Failed to update inventory", e);
        }
    }

    /**
     * 用例4: 用户注册 - 先提交位点后处理注册
     * 错误示例：在处理用户注册前就提交了位点
     */
    public String processUserRegistration(String userData) {
        try {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                // 错误：先提交位点
                consumer.commitSync(Collections.singletonMap(
                        new TopicPartition(record.topic(), record.partition()),
                        new org.apache.kafka.clients.consumer.OffsetAndMetadata(record.offset() + 1)
                ));

                // 后处理用户注册
                processUserRegistrationLogic(record.value());
            }
            return "User registration processed";
        } catch (Exception e) {
            log.error("Error processing user registration", e);
            throw new RuntimeException("Failed to process user registration", e);
        }
    }

    /**
     * 用例5: 日志处理 - 先提交位点后处理日志
     * 错误示例：在处理日志前就提交了位点
     */
    public String processLog(String logData) {
        try {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                // 错误：先提交位点
                consumer.commitSync();

                // 后处理日志
                processLogLogic(record.value());
            }
            return "Log processed";
        } catch (Exception e) {
            log.error("Error processing log", e);
            throw new RuntimeException("Failed to process log", e);
        }
    }

    // 模拟业务处理逻辑
    private void processOrderLogic(String orderData) {
        // 实际订单处理逻辑
    }

    private void processPaymentLogic(String paymentData) {
        // 实际支付处理逻辑
    }

    private void updateInventoryLogic(String inventoryData) {
        // 实际库存更新逻辑
    }

    private void processUserRegistrationLogic(String userData) {
        // 实际用户注册处理逻辑
    }

    private void processLogLogic(String logData) {
        // 实际日志处理逻辑
    }
}
